#include "MqInvoker.h"
#include "sds.h"
#include <stdio.h>
#include <time.h>
#include <errno.h>
#include <uthash.h>
#include "c11code.h"
#include "rwlock.h"
#include <debug_printf.h>

#ifdef WITH_IF_UTILITS_C
#include <if_utilits_c.h>
/** 获取设备 MAC 地址 */
static const char *get_mac_address()
{
    static char mac[32] = {0};
    if (mac[0] == '\0')
    {
        ifu_get_mac_address(nullptr, mac, sizeof(mac), true);
    }
    return mac;
}
/** 获取默认的 ack 通道名 */
/* 格式为 mqrpc_ack_14755B74C907, 其中 14755B74C907 为当设备 MAC 地址 */
/* 该函数用于生成 ack 通道名, 并在发送请求时使用 */ 
const char* thmar_default_get_ack_channel()
{
    static char ackChannel[64] = {'\0'};
    if (ackChannel[0] == '\0')
    {
        snprintf(ackChannel, sizeof(ackChannel), "mqrpc_ack_%s",get_mac_address());
    }
    return ackChannel;
}
#else
/** 获取当前时间的 nanos */
static long getCurrentTimeNanos()
{
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    return ts.tv_sec * 1000000000 + ts.tv_nsec;
}
/* 返回ack通道名默认实现*/ 
/* 格式为 mqrpc_ack_123456789, 其中 123456789 为当前时间的 nanos */ 
/* 该函数用于生成 ack 通道名, 并在发送请求时使用 */ 
const char* thmar_default_get_ack_channel()
{
    static char ackChannel[64] = {'\0'};
    if (ackChannel[0] == '\0')
    {
        snprintf(ackChannel, sizeof(ackChannel), "mqrpc_ack_%lx", getCurrentTimeNanos());
    }
    return ackChannel;
}
#endif

/** 获取当前时间的 ISO 8601 格式 */
/** 格式为 yyyy-MM-ddTHH:mm:ssZ, 其中 yyyy-MM-dd 为年月日, HH:mm:ss 为时分秒 */
static sds getCurrentTimeISO8601()
{
    // 获取当前时间
    time_t now;
    time(&now);
    // 将当前时间转换为 UTC 时间
    struct tm *utcTime = gmtime(&now);
    // 格式化为 ISO 8601
    return sdscatprintf(sdsnew(nullptr), "%04d-%02d-%02dT%02d:%02d:%02dZ",
                        utcTime->tm_year + 1900, // 年份从1900开始
                        utcTime->tm_mon + 1,     // 月份从0开始
                        utcTime->tm_mday,
                        utcTime->tm_hour,
                        utcTime->tm_min,
                        utcTime->tm_sec);
}


// RPC调用消息封装
// 包含 ack 通道名, 消息发送时间, RPC请求/服务端响应数据
// 该类用于封装 RPC 请求/响应数据, 并提供序列化/反序列化接口
class Envelope
{
public:
    sds ack;
    /** 消息发送时间 ISO861格式 */
    sds sendTime;
    /** 对于异常，是否只输出简要错误信息 */
    bool thinError;
    /** [JSON] RPC请求/服务端响应数据 */
    sds body;
    cJSON *bodyArray;
    Envelope()
        : ack(nullptr), sendTime(nullptr), body(nullptr), bodyArray(nullptr), thinError(false)
    {
    }
    ~Envelope()
    {
        sdsfree(ack);
        sdsfree(sendTime);
        sdsfree(body);
        cJSON_Delete(bodyArray);
    }
    Envelope(const Envelope &other)
    {
        this->ack = sdsnew(other.ack);
        this->sendTime = sdsnew(other.sendTime);
        this->body = sdsnew(other.body);
        this->bodyArray = cJSON_Duplicate(other.bodyArray, true);
    }
    Envelope(Envelope &&other)
    {
        this->ack = other.ack;
        this->sendTime = other.sendTime;
        this->body = other.body;
        this->bodyArray = other.bodyArray;
        other.ack = nullptr;
        other.sendTime = nullptr;
        other.body = nullptr;
        other.bodyArray = nullptr;
    }

    Envelope &operator=(Envelope &&other)
    {
        if (this != &other)
        {
            sdsfree(this->ack);
            sdsfree(this->sendTime);
            sdsfree(this->body);
            cJSON_Delete(this->bodyArray);
            this->ack = other.ack;
            this->sendTime = other.sendTime;
            this->body = other.body;
            this->bodyArray = other.bodyArray;
            other.ack = nullptr;
            other.sendTime = nullptr;
            other.body = nullptr;
            other.bodyArray = nullptr;
        }
        return *this;
    }
    Envelope &operator=(const Envelope &other)
    {
        if (this != &other)
        {
            sdsfree(this->ack);
            sdsfree(this->sendTime);
            sdsfree(this->body);
            cJSON_Delete(this->bodyArray);
            this->ack = sdsnew(other.ack);
            this->sendTime = sdsnew(other.sendTime);
            this->body = sdsnew(other.body);
            this->bodyArray = cJSON_Duplicate(other.bodyArray, true);
        }
        return *this;
    }
    // 解析 body 字符串, 获取 body 数组, 如果 body 为空则返回 nullptr
    cJSON *getBodyArray()
    {
        if (bodyArray == nullptr)
        {
            bodyArray = cJSON_Parse(body);
        }
        return bodyArray;
    }
    void setBodyArray(cJSON *bodyArray)
    {
        cJSON_Delete(this->bodyArray);
        this->bodyArray = bodyArray;
    }
    // 获取请求数组(bodyArray)中的第index个元素, 如果body为nullptr或index越界, 返回nullptr
    cJSON *getItemInArray(int index)
    {
        cJSON *array = getBodyArray();
        if (array == nullptr)
        {
            return nullptr;
        }
        auto size = cJSON_GetArraySize(array);
        if (index < 0 || index >= size)
        {
            return nullptr;
        }
        return cJSON_GetArrayItem(array, index);
    }
    // 获取seqid(消息序列号),
    // 即请求数组(bodyArray)中的第3个元素,如果body为nullptr或index越界, 返回-1
    // 参见 org.apache.thrift.protocol.TMessage.seqid
    int seqid()
    {
        cJSON *item = getItemInArray(3);
        if (item == nullptr)
        {
            return -1;
        }
        return item->valueint;
    }
    // 获取消息类型(type),
    // 参见 org.apache.thrift.protocol.TMessageType
    // 1: call, 2: reply, 3: exception, 4: oneway
    // 即请求数组(bodyArray)中的第2个元素, 如果body为nullptr或index越界, 返回-1
    int type()
    {
        cJSON *item = getItemInArray(2);
        if (item == nullptr)
        {
            return -1;
        }
        return item->valueint;
    }
    // 获取消息名称，即请求的方法名(name),
    // 参见 org.apache.thrift.protocol.TMessage.name
    // 即请求数组(bodyArray)中的第1个元素, 如果body为nullptr或index越界, 返回nullptr
    const char *name()
    {
        cJSON *item = getItemInArray(1);
        if (item == nullptr)
        {
            return nullptr;
        }
        return item->valuestring;
    }
    // 获取请求参数内容,
    // 即请求数组(bodyArray)中的第4个元素, 如果body为nullptr或index越界, 返回nullptr
    // detached : 为true从数组中分离出参数, 否则直接返回参数
    cJSON *content(bool detached = false)
    {
        if (detached)
        {
            return cJSON_DetachItemFromArray(getBodyArray(), 4);
        }
        return getItemInArray(4);
    }
    void setAck(const char *ack)
    {
        sdsfree(this->ack);
        this->ack = sdsnew(ack);
    }
    void setSendTime(const char *sendTime)
    {
        sdsfree(this->sendTime);
        this->sendTime = sdsnew(sendTime);
    }
    void setBody(const char *body)
    {
        sdsfree(this->body);
        this->body = sdsnew(body);
    }
};
// 将 Envelope 序列化为 json 字符串
// 返回0表示成功, 否则返回错误码
// 'json' 参数保存序列化后的 json 字符串
// 格式为: {"ack": "mqrpc_ack_123456789", "sendTime": "2021-01-01T00:00:00Z", "body": "json string"}
static int envelopeToJson(sds *json, const Envelope &envelope)
{
    cJSON *root = cJSON_CreateObject();
    if (!root)
    {
        return THMR_JSON_CREATE_OBJECT_FAILED;
    }
    cJSON_AddStringToObject(root, "ack", envelope.ack);
    cJSON_AddStringToObject(root, "sendTime", envelope.sendTime);
    cJSON_AddBoolToObject(root, "thinError", envelope.thinError);
    cJSON_AddStringToObject(root, "body", envelope.body);
    char* jsonStr = cJSON_PrintUnformatted(root);
    *json = sdsnew(jsonStr);
    COM_PRINTF_DEBUG("json=%s\n", *json);
    free(jsonStr);
    cJSON_Delete(root);
    return THMR_OK;
}
// 解析 json 字符串, 填充 Envelope 结构
// 返回0表示成功, 否则返回错误码
static int jsonToEnvelope(Envelope &envelope, const char *json, size_t length)
{
    cJSON *root = cJSON_ParseWithLength(json, length);
    if (!root)
    {
        return THMR_JSON_ERROR;
    }
    envelope.setAck(cJSON_GetObjectItem(root, "ack")->valuestring);
    envelope.setSendTime(cJSON_GetObjectItem(root, "sendTime")->valuestring);
    envelope.setBody(cJSON_GetObjectItem(root, "body")->valuestring);
    cJSON_Delete(root);
    return THMR_OK;
}

// 根据方法名和请求参数, 创建 Envelope 结构, 并设置 ack, sendTime, body 字段，
// 其中 ack 字段为 ack 通道名,
// sendTime 字段为当前时间的 ISO8601 格式字符串(自动生成),
// body 字段为请求参数(args)序列化后的 json 字符串
// 返回0表示成功, 否则返回错误码
static int createEnvelope(thmar::MqInvoker* mqInvoker, Envelope &envelope, const char *methodName, const cJSON *args)
{
    const char* ackChannel = mqInvoker->getAckChannel();
    if (!ackChannel)
    {
        return THMAR_NULL_ACK_CHANNEL;
    }
    envelope.setAck(ackChannel);
    envelope.thinError = mqInvoker->getOptions().thinError;
#ifndef NO_FILL_SEND_TIME
    auto ctstr = getCurrentTimeISO8601();
    envelope.setSendTime(ctstr);
    sdsfree(ctstr);
#endif
    cJSON *array = cJSON_CreateArray();
    if (!array)
    {
        return THMR_JSON_CREATE_ARRAY_FAILED;
    }
    cJSON_AddItemToArray(array, cJSON_CreateNumber(THRIFT_PROTOCOL_VERSION));
    cJSON_AddItemToArray(array, cJSON_CreateString(methodName));
    cJSON_AddItemToArray(array, cJSON_CreateNumber(THRIFT_MESSAGE_CALL)); // type
    cJSON_AddItemToArray(array, cJSON_CreateNumber(thmar_next_seqid()));  // seqid
    if (args)
    {
        cJSON_AddItemToArray(array, cJSON_Duplicate((cJSON *)args, true)); // args, 复制 args 避免内存泄漏
    }
    else
    {
        /* 如果 args 为 nullptr, 则创建一个空对象 */
        cJSON_AddItemToArray(array, cJSON_CreateObject());
    }
    char *jsonStr = cJSON_PrintUnformatted(array);
    envelope.setBody(jsonStr);
    free(jsonStr);
    envelope.setBodyArray(array);
    COM_PRINTF_DEBUG("envelope.body=%s\n", envelope.body);
    return THMR_OK;
}

class InvocationContext : public Envelope
{
public:
    void *responseContext;
    MQ_ONRESPONSE responseCallback;
    InvocationContext(const Envelope &envelope, void *responseContext, MQ_ONRESPONSE responseCallback)
        : Envelope(envelope), responseContext(responseContext), responseCallback(responseCallback)
    {
    }
};

////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////请求上下文哈希表管理//////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////

// 请求上下文
// 保存请求消息的序列号, 请求参数, 回调函数
// 用于在收到服务端响应消息时查找请求上下文, 并调用回调函数
// 该结构使用uthash实现哈希表, 用于保存请求上下文
// 哈希表的键为请求消息的序列号, 值为请求上下文指针
typedef struct RequestContextNode {
    int seqid;
    InvocationContext* context;
    UT_hash_handle hh;
    ~RequestContextNode()
    {
        delete context;
    }
}RequestContextNode;

/** 请求上下文哈希表 */ 
static RequestContextNode* requestContextHash = nullptr;
/** requestContextHash 读写锁 */
static rwlock_t requestContextHashLock = RWLOCK_INITIALIZER;

/** 初始化请求上下文哈希表 */
static void initRequestContextHash(){
    requestContextHash = nullptr;
    if(requestContextHashLock.initialized == 0){
        rwlock_init(&requestContextHashLock);
    }
}
/** 销毁请求上下文哈希表 */
static void destroyRequestContextHash(){
    if(requestContextHash!= nullptr){
        RequestContextNode* requestContext, *tmp;
        HASH_ITER(hh, requestContextHash, requestContext, tmp)
        {
            HASH_DEL(requestContextHash, requestContext);
            delete requestContext;
        }
        HASH_CLEAR(hh, requestContextHash);
        requestContextHash = nullptr;
    }
    if(requestContextHashLock.destroyed == 0){
        rwlock_destroy(&requestContextHashLock);
    }
}

/** 查找请求上下文 */ 
static RequestContextNode* findRequestContext(int seqid)
{
    rwlock_lock_read(&requestContextHashLock);
    RequestContextNode* requestContext = nullptr;
    HASH_FIND_INT(requestContextHash, &seqid, requestContext);
    rwlock_unlock_read(&requestContextHashLock);
    return requestContext;
}
/** 移除请求上下文 */
static void removeRequestContext(RequestContextNode* requestContext)
{
    rwlock_lock_write(&requestContextHashLock);
    HASH_DEL(requestContextHash, requestContext);
    delete requestContext;
    rwlock_unlock_write(&requestContextHashLock);
}
/** 添加请求上下文 */
static void addRequestContext(RequestContextNode* requestContext)
{
    rwlock_lock_write(&requestContextHashLock);
    HASH_ADD_INT(requestContextHash, seqid, requestContext);
    rwlock_unlock_write(&requestContextHashLock);
}
////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////


// 回调函数
// 处理服务端响应消息
// 解析消息, 并根据消息序列号查找请求上下文, 调用回调函数
// 移除请求上下文
static void onResponse(void *context, const char *message, size_t length)
{
    Envelope responseMessage;
    // 解析消息
    if (jsonToEnvelope(responseMessage, message, length))
    {
        COM_PRINTF_DEBUG("Failed to parse response message\n");
        return;
    }
    COM_PRINTF_DEBUG("responseMessage.seqid=%d\n", responseMessage.seqid());
    COM_PRINTF_DEBUG("responseMessage.body=%s\n", responseMessage.body);

    // 根据消息序列号查找请求上下文，并调用回调函数
    RequestContextNode* requestContext = findRequestContext(responseMessage.seqid());
    // 请求上下文不存在, 忽略该消息
    if(requestContext == nullptr){
        COM_PRINTF_DEBUG("request context not found\n");
        return;
    }

    // 检查requestContext->context是否为nullptr
    if(requestContext->context == nullptr){
        COM_PRINTF_DEBUG("request context is null\n");
        removeRequestContext(requestContext); // 移除请求上下文
        return;
    }

    // 调用回调函数
    requestContext->context->responseCallback(requestContext->context->responseContext, responseMessage.content(true));

    // 移除请求上下文
    removeRequestContext(requestContext);
}


thmar::MqInvoker::MqInvoker(smq::ThmarMqttClient *client, const MqInvoker_options &options) : options(options), client(client),ackChannel(nullptr)
{
    /** 初始化响应频道 */
    initAckChannel();
}

thmar::MqInvoker::~MqInvoker()
{
    delete client;
    client = nullptr;
    free((void*)ackChannel);
    ackChannel = nullptr;
}

void thmar::MqInvoker::initAckChannel()
{
    if (!ackChannel)
    {
        const char *_c;
        if (options.getAckChannel_f)
        {
            _c = options.getAckChannel_f();
        }
        else
        {
            _c = thmar_default_get_ack_channel();
        }
        if(_c == nullptr){
            com_debug_printf("get ack channel failed\n");
            return;
        }
        ackChannel = strdup(_c);
    }
}

const MqInvoker_options &thmar::MqInvoker::getOptions()
{
    return this->options;
}

int thmar::MqInvoker::asyncCall(const char *requestChannel, const char *methodName, const cJSON *args,void *responseContext, MQ_ONRESPONSE responseCallback)
{
    if (!requestChannel || !methodName || !responseCallback)
    {
        return THMR_NULL_POINTER;
    }
    int code;
    Envelope envelope;
    if (code = createEnvelope(this, envelope, methodName, args))
    {
        return code;
    }
    sds message;
    if (code = envelopeToJson(&message, envelope))
    {
        return code;
    }
    if (message == nullptr)
    {
        return THMR_JSON_PRINT_FAILED;
    }
    // 创建请求上下文
    RequestContextNode* requestContext = new RequestContextNode();
    requestContext->seqid = envelope.seqid();
    requestContext->context = new InvocationContext(envelope, responseContext, responseCallback);
    addRequestContext(requestContext);
    COM_PRINTF_DEBUG("publish request message to %s\nmessage=%s\n", requestChannel, message);
    // 发布请求消息
    code = client->publish(requestChannel, message);
    sdsfree(message);
    if (code)
    {
        COM_PRINTF_DEBUG("publish failed, code=%d\n", code);
        /* 发送失败, 移除请求上下文 */
        removeRequestContext(requestContext);
        return code;
    }
    return code;
}
// 同步器回调函数
// 异步调用后, 等待服务端响应, 并返回响应参数
// 该函数会阻塞当前线程, 直到收到服务端响应
struct Syncer{
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    bool done;
    /** 服务调用响应参数 */
    /* 参见 onSyncResponse */
    cJSON* response;
    Syncer()
    {
        pthread_mutex_init(&mutex, nullptr);
        pthread_cond_init(&cond, nullptr);
        done = false;
    }
    ~Syncer()
    {
        pthread_mutex_destroy(&mutex);
        pthread_cond_destroy(&cond);
    }
    void wait()
    {
        pthread_mutex_lock(&mutex);
        while (!done)
        {
            pthread_cond_wait(&cond, &mutex);
        }
        pthread_mutex_unlock(&mutex);
    }
    /* 等待指定时间, 超时则返回 */
    int wait(int timeoutMills)
    {
        struct timespec ts;
        clock_gettime(CLOCK_REALTIME, &ts);
        ts.tv_sec += timeoutMills / 1000;
        ts.tv_nsec += (timeoutMills % 1000) * 1000000;
        if (ts.tv_nsec >= 1000000000)
        {
            ts.tv_sec++;
            ts.tv_nsec -= 1000000000;
        }
        pthread_mutex_lock(&mutex);
        while (!done)
        {
            int ret = pthread_cond_timedwait(&cond, &mutex, &ts);
            if (ret == ETIMEDOUT)
            {
                break;
            }
        }
        pthread_mutex_unlock(&mutex);
        return done ? THMR_OK : THMR_TIMEOUT;
    }
    void notify()
    {
        pthread_mutex_lock(&mutex);
        done = true;
        pthread_cond_signal(&cond);
        pthread_mutex_unlock(&mutex);
    }
};

/* 同步器回调函数 */
static void onSyncResponse(void *context, cJSON* response)
{
    Syncer* syncer = (Syncer*)context;
    syncer->response = response;
    syncer->notify();
}

int thmar::MqInvoker::syncCall(const char* requestChannel,const char* methodName, const cJSON* args,cJSON** response)
{
    if (!requestChannel || !methodName || !response)
    {
        return THMR_NULL_POINTER;
    }
    /* 同步器 */
    Syncer syncer;
    /* 执行异步调用 */
    int code = asyncCall(requestChannel, methodName, args, &syncer, onSyncResponse);
    if (code)
    {
        COM_PRINTF_DEBUG("asyncCall failed, code=%d\n", code);
        return code;
    }
    /* 等待同步响应 */
    code = syncer.wait(options.requestTimeout*1000);
    if(code == THMR_OK){
        /* 复制响应参数 */ 
        *response = syncer.response;
        COM_PRINTF_DEBUG("syncCall success\n");
    }else{
        COM_PRINTF_DEBUG("syncCall timeout\n");
        *response = nullptr;
    }
    return code;
}

int thmar::MqInvoker::start()
{
    initRequestContextHash();
    int code = client->start();
    if (code)
    {
        return code;
    }
    code = client->registerChannel(getAckChannel(),this, onResponse);
    return code;
}

void thmar::MqInvoker::stop()
{
    client->stop();
    destroyRequestContextHash();
}

const char *thmar::MqInvoker::getAckChannel()
{
    return ackChannel;
}

void thmar::MqInvoker::printOptions()
{
    printf("options:\n");
    printf("  requestTimeout=%d\n", options.requestTimeout);
    printf("  ackChannel=%s\n", getAckChannel());
}

THMAR_API thmar_mq_invoker_t thmar_mq_invoker_create(thmar_mqtt_client_t client, const MqInvoker_options *options)
{
    if (!client || !options)
    {
        return nullptr;
    }
    thmar::MqInvoker *invoker = new thmar::MqInvoker((smq::ThmarMqttClient *)client, *options);
    return (thmar_mq_invoker_t)invoker;
}

THMAR_API void thmar_mq_invoker_destroy(thmar_mq_invoker_t invoker)
{
    if (invoker)
    {
        delete (thmar::MqInvoker *)invoker;
    }
}

THMAR_API int thmar_mq_invoker_async_call(thmar_mq_invoker_t invoker, const char *requestChannel, const char *methodName, const cJSON *args, void *responseContext, MQ_ONRESPONSE responseCallback)
{
    if (!invoker || !requestChannel || !methodName || !responseCallback)
    {
        return THMR_NULL_POINTER;
    }
    thmar::MqInvoker *mqInvoker = (thmar::MqInvoker *)invoker;
    return mqInvoker->asyncCall(requestChannel, methodName, args, responseContext, responseCallback);
}

THMAR_API int thmar_mq_invoker_sync_call(thmar_mq_invoker_t invoker, const char *requestChannel, const char *methodName, const cJSON *args, cJSON **response)
{
    if (!invoker || !requestChannel || !methodName || !response)
    {
        return THMR_NULL_POINTER;
    }
    thmar::MqInvoker *mqInvoker = (thmar::MqInvoker *)invoker;
    return mqInvoker->syncCall(requestChannel, methodName, args, response);
}

THMAR_API int thmar_mq_invoker_start(thmar_mq_invoker_t invoker)
{
    if (!invoker)
    {
        return THMR_NULL_POINTER;
    }
    thmar::MqInvoker *mqInvoker = (thmar::MqInvoker *)invoker;
    return mqInvoker->start();
}

THMAR_API void thmar_mq_invoker_stop(thmar_mq_invoker_t invoker)
{
    if (invoker)
    {
        thmar::MqInvoker *mqInvoker = (thmar::MqInvoker *)invoker;
        mqInvoker->stop();
    }
}

THMAR_API void thmar_mq_invoker_print_options(thmar_mq_invoker_t invoker)
{
    if (invoker)
    {
        thmar::MqInvoker *mqInvoker = (thmar::MqInvoker *)invoker;
        mqInvoker->printOptions();
    }
}